package org.hong.monkey.rpc

import org.hong.monkey.util.RpcUtils
import org.hong.monkey.{MonkeyException, Logging, MonkeyConf}

import scala.concurrent.Future
import scala.reflect.ClassTag

/**
 */
private[monkey] abstract class RpcEndpointRef(conf: MonkeyConf)
  extends Serializable with Logging {

  private[this] val maxRetries = RpcUtils.numRetries(conf)
  private[this] val retryWaitMs = RpcUtils.retryWaitMs(conf)
  private[this] val defaultAskTimeout = RpcUtils.askRpcTimeout(conf)

  def address: RpcAddress

  def name: String

  def send(message: Any): Unit

  def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T]

  def ask[T: ClassTag](message: Any): Future[T] = ask(message, defaultAskTimeout)

  def askWithRetry[T: ClassTag](message: Any): T = askWithRetry(message, defaultAskTimeout)

  def askWithRetry[T: ClassTag](message: Any, timeout: RpcTimeout): T = {
    var attempts = 0
    var lastException: Exception = null
    while (attempts < maxRetries) {
      attempts += 1
      try {
        val future = ask[T](message, timeout)
        val result = timeout.awaitResult(future)
        if (result == null)
          throw new MonkeyException("RpcEndpoint returned null")
        return result
      } catch {
        case ie: InterruptedException => throw ie
        case e: Exception =>
          lastException = e
          logWarning(s"Error sending message [message = $message] in $attempts attempts", e)
      }

      if (attempts < maxRetries)
        Thread.sleep(retryWaitMs)
    }

    throw new MonkeyException(s"Error sending message [message = $message]", lastException)
  }

}
